/*
 * Copyright (c) 2021 IBM Corp and others.
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v2.0
 * and Eclipse Distribution License v1.0 which accompany this distribution.
 *
 * The Eclipse Public License is available at
 *    https://www.eclipse.org/legal/epl-2.0/
 * and the Eclipse Distribution License is available at
 *   http://www.eclipse.org/org/documents/edl-v10.php.
 *
 * Contributors:
 *    Seth Hoenig
 *    Allan Stockdill-Mander
 *    Mike Robertson
 */

package main

import (
	"fmt"
	"github.com/eclipse/paho.mqtt.golang"
	"os"
	"time"
)

var count int = 500
var record int

var f mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
	fmt.Println(time.Now().String())
	fmt.Println("topic： ", msg.Topic())
	fmt.Println("收到数据长度: ", len(msg.Payload()))
	record++
	fmt.Println(record)
}

func main() {
	//mqtt.DEBUG = log.New(os.Stdout, "", 0)
	//mqtt.ERROR = log.New(os.Stdout, "", 0)
	opts := mqtt.NewClientOptions().AddBroker("tcp://192.168.20.80:1883")

	c := mqtt.NewClient(opts)
	defer func() {
		fmt.Println("断开连接")
		c.Disconnect(250)
	}()
	token := c.Connect()
	if token.Wait() && token.Error() != nil {
		panic(token.Error())
	}

	go func() {
		if token := c.Subscribe("qtmd", 0, f); token.Wait() && token.Error() != nil {
			fmt.Println(token.Error())
			os.Exit(1)
		}
	}()

	b := make([]byte, 0)
	for i := 0; i < 1024 * 1024; i++ {
		b = append(b, 255)
	}
	time.Sleep(time.Second * 2)
	go func(b []byte) {
		for {
			go func() {
				token := c.Publish("qtmd", 0, false, b)
				token.Wait()
			}()
			time.Sleep(time.Second)
		}
	}(b)

	for {
		time.Sleep(time.Second)
	}
}
